/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

import jvs.peer.util.Log;


/**
 * @author qiangli
 *
 */
public class MessageInputStream extends PipedInputStream {
	private Channel channel;
	private PipedOutputStream out;
	private int seqno = 0; //messages received
	private int counter = 0; //bytes received
	
	public MessageInputStream(Channel channel) throws IOException {
		this.channel = channel;
		this.out = new PipedOutputStream(this);
	}
	
	synchronized public void receive(Message msg) throws IOException {
		final int ioscmd = msg.getIoscmd();
		
		if (msg.getSeqno() != seqno) {
			throw new IOException("Message out of sequence, expecting:  "+seqno+" received: "+msg.getSeqno()+" channel: "+channel.getId());
		}
		seqno++;
		
		switch(ioscmd) {
		case MessageConstant.open:
			break;
		case MessageConstant.data:
			byte[] buf = msg.getData();
			int size = msg.getSize();
			counter += size;
			
			out.write(buf, 0, size);
			out.flush();
			break;
		case MessageConstant.close:
			close();
			break;
		default:
			if (msg.getSeqno() != seqno) {
				throw new IOException("Invalid ioscmd: "+ ioscmd + " channel: "+channel.getId());
			}
		}
		
		if (Log.DEBUG) {
			Log.log(this, "ioscmd: "+ ioscmd + " "+Thread.currentThread().getId()+msg.getSeqno());
		}
	}

	@Override
	synchronized public void close() throws IOException {
		if (out != null) {
			out.flush();

			super.close();
			
			out.close();
		}
		out = null;
		channel = null;
	}

}
